package com.roy.rocketmq.basic;

import com.alibaba.fastjson.JSON;
import com.roy.rocketmq.domain.OrderPaidEvent;
import com.roy.rocketmq.domain.User;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.MimeTypeUtils;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * @author yuanjie.Deng
 * @version 1.0
 * @date 2022/8/10 09:42
 */
@Component
public class MyProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    private static final String springTopic = "TestTopic";


    /**
     * 发送同步字符串消息
     */
    public SendResult syncSendStr() {
        //发送字符串消息  springTopic: 主题  payload: 消息
        SendResult sendResult = this.rocketMQTemplate.syncSend(springTopic, "Hello, World!");
        System.out.printf("syncSend1 to topic %s sendResult=%s %n", springTopic, sendResult);
        return sendResult;
    }

    /**
     * 发送同步Object消息
     */
    public void syncSendObject() {
        // 发送对象
        SendResult sendResult = rocketMQTemplate.syncSend(springTopic, new User().setUserAge((byte) 18).setUserName("DDD"));
        System.out.printf("syncSend2 to topic %s sendResult=%s %n", springTopic, JSON.toJSONString(sendResult));
    }


    /**
     * 发送同步Object消息
     */
    public void syncSendMessage() {

        SendResult sendResult = rocketMQTemplate.syncSend(springTopic, MessageBuilder.withPayload(
                new User().setUserAge((byte) 21).setUserName("Lester")).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build());
        System.out.printf("syncSend3 to topic %s sendResult=%s %n", springTopic, JSON.toJSONString(sendResult));
    }

    /**
     * 发送异步消息
     */
    public void asyncSend() {

        //发送对象消息
        rocketMQTemplate.asyncSend(springTopic, new User().setUserAge((byte) 18).setUserName("DDD11"), new SendCallback() {
            // 成功
            @Override
            public void onSuccess(SendResult var1) {
                System.out.printf("async onSucess SendResult=%s %n", var1);
            }

            // 失败
            @Override
            public void onException(Throwable var1) {
                System.out.printf("async onException Throwable=%s %n", var1);
            }
        });
    }


    /**
     * 发送指定TAG的消息
     */
    public void convertAndSend() {
        rocketMQTemplate.convertAndSend(springTopic + ":ddd", "I'm from tag0");
        System.out.printf("syncSend topic %s tag %s %n", springTopic, "tag0");
    }


    /**
     * 同步发送消息并且等待消费者回执信息
     */
    public void syncSendAndReceive() {


        //同步发送消息并且返回一个String类型的结果。
        String replyString = rocketMQTemplate.sendAndReceive("ddd", "request string", String.class);
        System.out.printf("send %s and receive %s %n", "request string", replyString);
        //同步发送消息并且返回一个Byte数组类型的结果。
        byte[] replyBytes = rocketMQTemplate.sendAndReceive(springTopic, MessageBuilder.withPayload("request byte[]").build(), byte[].class, 3000);
        System.out.printf("send %s and receive %s %n", "request byte[]", new String(replyBytes));
//
//        //同步发送一个带hash参数的请求(排序消息)，并返回一个User类型的结果
//        User requestUser = new User().setUserAge((byte) 9).setUserName("requestUserName");
//        User replyUser = rocketMQTemplate.sendAndReceive(springTopic, requestUser, User.class, "order-id");
//        System.out.printf("send %s and receive %s %n", requestUser, replyUser);
//
//        //同步发送一个带延迟级别的消息(延迟消息)，并返回一个泛型结果
//        ProductWithPayload<String> replyGenericObject = rocketMQTemplate.sendAndReceive(springTopic, "request generic",
//                new TypeReference<ProductWithPayload<String>>() {
//                }.getType(), 30000, 2);
//        System.out.printf("send %s and receive %s %n", "request generic", replyGenericObject);
    }

    /**
     * 异步发送消息
     */
    public void asyncSendAndReceive() {


        //异步发送消息，返回String类型结果。
        rocketMQTemplate.sendAndReceive("ddd", "request string", new RocketMQLocalRequestCallback<String>() {
            @Override
            public void onSuccess(String message) {
                System.out.printf("send %s and receive %s %n", "request string", message);
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        });
//        //异步发送消息，并返回一个User类型的结果。
//        rocketMQTemplate.sendAndReceive(springTopic, new User().setUserAge((byte) 9).setUserName("requestUserName"), new RocketMQLocalRequestCallback<User>() {
//            @Override
//            public void onSuccess(User message) {
//                System.out.printf("send user object and receive %s %n", message.toString());
//            }
//
//            @Override
//            public void onException(Throwable e) {
//                e.printStackTrace();
//            }
//        }, 5000);
    }

    /**
     * 发送批量消息
     */
    public void sendBatch() {
        //发送批量消息
        List<Message> msgs = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            msgs.add(MessageBuilder.withPayload("Hello RocketMQ Batch Msg#" + i).
                    setHeader(RocketMQHeaders.KEYS, "KEY_" + i).build());
        }
        SendResult sr = rocketMQTemplate.syncSend(springTopic, msgs, 60000);
        System.out.printf("--- Batch messages send result :" + sr);
    }


    /**
     * 发送批量消息
     */
    public void sendScheduled() {
        //发送批量消息

        SendResult sr = rocketMQTemplate.syncSend(springTopic, MessageBuilder.withPayload("Scheduled").build(), 3000L, 3);
        System.out.println("sendScheduled messages send result :" + sr);
    }


    /**
     * 发送事务消息
     */
    public void sendTransaction() {
        //发送批量消息

        String transactionId = UUID.randomUUID().toString();
        //如果可以删除订单则发送消息给rocketmq，让用户中心消费消息
        OrderPaidEvent orderPaidEvent = new OrderPaidEvent();
        orderPaidEvent.setOrderId("11");
        orderPaidEvent.setPaidMoney(new BigDecimal("112"));
        rocketMQTemplate.sendMessageInTransaction("add-amount",
                MessageBuilder.withPayload(
                                orderPaidEvent
                        )
                        .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
                        .setHeader("order_id", orderPaidEvent.getOrderId())
                        .build()
                , orderPaidEvent
        );
    }


}
